package com.wei.violet.app.media


import com.wei.violet.bean.AppDataCase
import com.wei.violet.config.ConfigHandle
import com.wei.violet.utils.BaseDataKpi
import org.apache.commons.lang.StringUtils
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}
import org.apache.spark.{SparkConf, SparkContext}

object SparkCoreMediaHandler {


  Logger.getLogger("org").setLevel(Level.WARN)
  def main(args: Array[String]): Unit = {


    val conf: SparkConf = new SparkConf()
      .setAppName("将终端设备统计结果添加到mysql数据库中")
      .setMaster("local[*]")
      .set("spark.serizlizer", "org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(conf)
    //读取广播数据
    val appdict=sc.textFile("F:\\bigdataFile\\data\\app_dict.txt")

    val appmapped = appdict.map(_.split("\\t",-1)).filter(_.length>=5).map(str => {
      (str(4),str(1))
    })

    val arrApp = appmapped.collectAsMap()
    // 把规则库的数据进行广播
    val bc = sc.broadcast(arrApp)

    val sQLContext = new SQLContext(sc)
    val dataFrame: DataFrame = sQLContext.read.parquet("F:\\bigdataFile\\parquet")

    val basedata = dataFrame.map(d => {
         var appname=d.getAs[String]("appname")
         val appid=d.getAs[String]("appid")

          if(StringUtils.isEmpty(appname)){
            if(StringUtils.isNotEmpty(appid)){
              appname=bc.value.getOrElse(appid,appid)
            }else{

              appname="未知"
            }
          }

      (appname,BaseDataKpi(d))
    })

      val media=basedata.map(d=>((d._1),(d._2))).reduceByKey((a, b) => a.zip(b).map(t => t._1 + t._2))
    import sQLContext.implicits._
    val result = media.map(x =>(AppDataCase(
      x._1, x._2(0).toInt, x._2(1).toInt, x._2(2).toInt, x._2(3).toInt, x._2(4).toInt, x._2(5).toInt, x._2(6).toInt, x._2(7), x._2(8))
    )).toDF()

    result.write.mode(SaveMode.Overwrite).jdbc(
      ConfigHandle.url,
      "sparkcoreMedia",
      ConfigHandle.conn
    )

    sc.stop()

  }

}
